package com.xinqi.common.rocketmq.processor;

import com.xinqi.common.base.utils.json.JSON;
import com.xinqi.common.rocketmq.util.ByteUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

/**
 * @author linkai
 * @date 2023/11/19
 */
@Slf4j
public class MqConsumerListener<T> implements MessageListener {
    private final Object target;

    private final Method method;

    private final Class<T> typeClass;

    public MqConsumerListener(Object target, Method method, Class<T> typeClass) {
        this.target = target;
        this.method = method;
        this.typeClass = typeClass;
    }

    @Override
    public ConsumeResult consume(MessageView messageView) {
        try {
            T body = JSON.parse(ByteUtils.byteBufferToStr(messageView.getBody()), typeClass);
            Object[] params = new Object[method.getParameterTypes().length];
            for (int i = 0; i < method.getParameterTypes().length; i++) {
                Class<?> parameterType = method.getParameterTypes()[i];
                if (parameterType.isAssignableFrom(typeClass)) {
                    params[i] = body;
                } else if (parameterType.isAssignableFrom(MessageView.class)) {
                    params[i] = messageView;
                }
            }
            ReflectionUtils.makeAccessible(this.method);
            Object invoke = this.method.invoke(this.target, params);
            Class<?> returnType = method.getReturnType();
            if (returnType.isAssignableFrom(Void.class)) {
                return ConsumeResult.SUCCESS;
            } else if (returnType.isAssignableFrom(ConsumeResult.class)) {
                return (ConsumeResult) invoke;
            }
        } catch (InvocationTargetException ex) {
            ReflectionUtils.rethrowRuntimeException(ex.getTargetException());
        } catch (IllegalAccessException ex) {
            log.error("消息消费失败，方法没有访问权限", ex);
            return ConsumeResult.FAILURE;
        }
        return ConsumeResult.SUCCESS;
    }
}
